
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：test.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/5/24 16:56
'''

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
from datetime import datetime

'''
@Project ：bignet_sumo_dev
@File ：main-dev-tongluoshan.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/3/24 9:40
'''


import math
import queue
import json
import threading
import random
import time
from multiprocessing import Process, Queue

import traci.constants as tc
import sumolib

from kafka import KafkaProducer, KafkaConsumer
from pyproj import Proj


from utils import sumo_utils
from loguru import logger
from utils.tools import calculate_distance, find_vehicles_by_duration_optimized, \
    calculate_vehicle_acceleration_deceleration, get_timestamp_milliseconds
import concurrent.futures
kafka_producer = KafkaProducer(bootstrap_servers='10.100.12.58:9093', key_serializer=str.encode,
                               value_serializer=lambda x: json.dumps(x).encode('utf-8'))
RADAR_DATA_LIST = {
                      33: {'sn': 33, 'lon': 113.821658112, 'lat': 34.562413628, 'Downstream': {'sn': 35},
                           'Upstream': None,
                           'disDown': 2393, 'normal': True},
                      35: {'sn': 35, 'lon': 113.815275031, 'lat': 34.541538249, 'Downstream': {'sn': 37},
                           'Upstream': {'sn': 33},
                           'disDown': 1145, 'normal': True},
                      37: {'sn': 37, 'lon': 113.810490264, 'lat': 34.532017004, 'Downstream': {'sn': 17},
                           'Upstream': {'sn': 35},
                           'disDown': 363, 'normal': True},
                      17: {'sn': 17, 'lon': 113.809019865, 'lat': 34.528977466, 'Downstream': {'sn': 19},
                           'Upstream': {'sn': 37},
                           'disDown': 593, 'normal': True},
                      19: {'sn': 19, 'lon': 113.806568569, 'lat': 34.524037319, 'Downstream': {'sn': 21},
                           'Upstream': {'sn': 17},
                           'disDown': 596, 'normal': True},
                      21: {'sn': 21, 'lon': 113.804643991, 'lat': 34.518910304, 'Downstream': {'sn': 23},
                           'Upstream': {'sn': 19},
                           'disDown': 324, 'normal': True},
                      23: {'sn': 23, 'lon': 113.804028033, 'lat': 34.516033499, 'Downstream': {'sn': 25},
                           'Upstream': {'sn': 21},
                           'disDown': 777, 'normal': True},
                      25: {'sn': 25, 'lon': 113.802948288, 'lat': 34.509094823, 'Downstream': {'sn': 27},
                           'Upstream': {'sn': 23},
                           'disDown': 578, 'normal': True},
                      27: {'sn': 27, 'lon': 113.803285954, 'lat': 34.503899314, 'Downstream': {'sn': 29},
                           'Upstream': {'sn': 25},
                           'disDown': 677, 'normal': True},
                      29: {'sn': 29, 'lon': 113.803775624, 'lat': 34.497822559, 'Downstream': {'sn': 31},
                           'Upstream': {'sn': 27},
                           'disDown': 1018, 'normal': True},
                      31: {'sn': 31, 'lon': 113.804504153, 'lat': 34.488680039, 'Downstream': None,
                           'Upstream': {'sn': 29}, 'normal': True},

}
def time_str(timestamp_in_milliseconds):
    # 转换为datetime对象
    date_time_obj = datetime.fromtimestamp(timestamp_in_milliseconds / 1000)
    # 格式化输出为你想要的格式
    formatted_time = date_time_obj.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

    return formatted_time
logger.add("debug1.log", format="{time} {level} {message}", level="DEBUG")
# 高密度
# file_path = r"F:\辛英杰\proj2025\河南数字换转型\kafkatool\export\20250528-090000_20250528-110000.json"
# 中密度
file_path = r"F:\辛英杰\proj2025\河南数字换转型\kafkatool\export\20250528-230000_20250529-010000.json"
# 低密度
# file_path = r"F:\辛英杰\proj2025\河南数字换转型\kafkatool\export\20250529-020000_20250529-040000.json"
with open(file_path,'r',encoding='utf-8') as f:
    while True:
        line = f.readline().strip()
        if line:
            data = json.loads(line)

            data['globalTimeStamp'] = time_str(int(time.time() * 1000))
            kafka_producer.send('e1_fusion_data_v8', value=data,
                                key=str(123))
            time.sleep(0.1)
        else:
            break

